#pragma  once
#include <vector>
#include <iostream>
#include <cassert>
#include <string>
#include <cstring>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <unistd.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <sys/eventfd.h>
#include <sys/timerfd.h>

#define INF 0
#define DBG 1
#define ERR 2
#define DEFAULT_LOG DBG

#define LOG(level, format, ...)                                                            \
    do                                                                                     \
    {                                                                                      \
        if (level < DEFAULT_LOG)                                                           \
            break;                                                                         \
        time_t t = time(0);                                                                \
        struct tm *ltm = localtime(&t);                                                    \
        char tmp[32] = {0};                                                                \
        strftime(tmp, 31, "%H:%M:%S", ltm);                                                \
        fprintf(stdout, "[%p %s %s:%d]" format "\n",(void*)pthread_self(),  tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define ILOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DLOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

namespace wyl
{
    // 缓冲区类
#define BUFFER_DEFAULT_SIZE 1024
    class Buffer
    {
    private:
        std::vector<char> _buffer; // 缓冲区
        uint64_t _reader_idx;      // 读取的位置
        uint64_t _writer_idx;      // 写入的位置
    public:
        Buffer() : _buffer(BUFFER_DEFAULT_SIZE), _reader_idx(0), _writer_idx(0) {}

    public:
        char *Begin() { return &*_buffer.begin(); }
        // 获取当前写入的位置
        char *WritePosition() { return Begin() + _writer_idx; }
        // 获取当前读取位置
        char *ReadPosition() { return Begin() + _reader_idx; }

        // 获取0 - 读取位置的大小
        uint64_t HeadIdleSize() { return _reader_idx; }

        // 获取写入位置到_buffer.size()位置的大小
        uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }

        // 获取可读取数据的大小
        uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }

        // 将读取偏移向后移动
        void MoveReadOffset(uint64_t len)
        {
            if(len == 0) return;
            assert(len <= ReadAbleSize());
            _reader_idx += len;
        }

        // 写入偏移向后移动
        void MoveWriteOffset(uint64_t len)
        {
            if(len == 0) return;
            assert(len <= TailIdleSize());
            _writer_idx += len;
        }

        // 确保写入数据时空间足够
        void EnsureWriteSpace(uint64_t len)
        {
            // 如果写入位置后有足够的空间，不需要扩容
            if (len <= TailIdleSize())
                return;

            // 如果数组剩余的所有空间大于len，那么把read往前移
            if (len <= TailIdleSize() + HeadIdleSize())
            {
                uint64_t rsz = ReadAbleSize();
                std::copy(ReadPosition(), WritePosition(), _buffer.begin()); // 把read_idx-write_idx区间移动到数组的开头
                _reader_idx = 0;
                _writer_idx = rsz;
            }
            else
            {
                // len > TailIdleSize() + HeadIdleSize()的情况，直接扩容write_idx + len个即可
                _buffer.resize(_writer_idx + len);
            }
        }

        // 把给的缓冲区写入进来
        void Write(const void *data, uint64_t len)
        {
            if (len == 0)
                return;
            EnsureWriteSpace(len);
            std::copy((char *)data, (char *)data + len, WritePosition());
        }

        void WriteAndPush(const void *data, uint64_t len)
        {
            Write(data, len);
            MoveWriteOffset(len);
        }

        // 把String对象的数据写入进来
        void WriteString(const std::string &data, uint64_t len)
        {
            Write(data.c_str(), len);
        }
        void WriteStringAndPush(const std::string &data, uint64_t len)
        {
            WriteString(data, len);
            MoveWriteOffset(len);
        }

        // 把Buffer对象的数据写入进来
        void WriteBuffer(Buffer &buf)
        {
            Write(buf.ReadPosition(), buf.ReadAbleSize());
        }
        void WriteBufferAndPush(Buffer &buf)
        {
            WriteBuffer(buf);
            MoveWriteOffset(buf.ReadAbleSize());
        }

        // 读取缓冲区到用户给的缓冲区
        void Read(void *buf, uint64_t len)
        {
            assert(len <= ReadAbleSize());
            std::copy(ReadPosition(), ReadPosition() + len, (char *)buf);
        }
        void ReadAndPop(void *buf, uint64_t len)
        {
            Read(buf, len);
            MoveReadOffset(len);
        }

        // 读取缓冲区到字符串
        void ReadAsString(std::string *out, uint64_t len)
        {
            assert(len <= ReadAbleSize());
            out->resize(len); // 保证out的空间足够
            Read(&(*out)[0], len);
        }

        void ReadAsStringAndPop(std::string *out, uint64_t len)
        {
            ReadAsString(out, len);
            MoveReadOffset(len);
        }

        // 找到字符串中换行所在的地址
        char *FindCRLF()
        {
            return (char *)memchr(ReadPosition(), '\n', ReadAbleSize());
        }

        // 提取一行
        std::string GetLine()
        {
            char *crlf = FindCRLF();
            if (crlf == nullptr)
                return "";
            std::string str;
            // +1是为了把换行符读进来
            ReadAsString(&str, crlf + 1 - ReadPosition());
            ILOG("提取一行：%s",str.c_str());
            ILOG("%s--------------",ReadPosition());
            return str;
        }
        // 提取一行并清除提取行的缓冲区
        std::string GetLineAndPop()
        {
            std::string str = GetLine();
            MoveReadOffset(str.size());
            return str;
        }

        // 清空缓冲区
        void Clear()
        {
            _reader_idx = _writer_idx = 0;
        }
    };

// 套接字类
#define MAX_LISTEN 1024
    class Socket
    {
    public:
        Socket(int sock = -1) : _sock(sock) {}

    public:
        int GetFd() { return _sock; }
        bool CreateSocket()
        {
            _sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
            if (_sock < 0)
            {
                ELOG("socket created failed");
                return false;
            }
            return true;
        }

        bool Bind(const std::string &ip, uint16_t port)
        {
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            addr.sin_port = htons(port);

            int ret = bind(_sock, (struct sockaddr *)&addr, sizeof(addr));
            if (ret < 0)
            {
                ELOG("socket bind failed");
                return false;
            }
            return true;
        }

        bool Listen(int backlog = MAX_LISTEN)
        {
            int ret = listen(_sock, backlog);
            if (ret < 0)
            {
                ELOG("socket listen failed");
                return false;
            }
            return true;
        }

        bool Connect(const std::string &ip, uint16_t port)
        {
            struct sockaddr_in svr;
            svr.sin_addr.s_addr = inet_addr(ip.c_str());
            svr.sin_family = AF_INET;
            svr.sin_port = htons(port);

            int ret = connect(_sock, (struct sockaddr *)&svr, sizeof(svr));
            if (ret < 0)
            {
                ELOG("socket connect failed");
                return false;
            }
            return true;
        }

        int Accept()
        {
            struct sockaddr_in client;
            socklen_t len;
            int ret = accept(_sock, (struct sockaddr *)&client, &len);
            if (ret < 0)
            {
                ELOG("socket Accept failed");
                return -1;
            }
            return ret;
        }

        ssize_t Recv(void *buf, size_t len, int flag = 0)
        {
            int ret = recv(_sock, buf, len, flag);
            // 阻塞模式下ret = 0 说明对端连接关闭力扣
            if (flag == 0 && ret == 0)
            {
                ELOG("Connection opposite is closed");
                return -1;
            }
            if (ret <= 0)
            {
                // 非阻塞模式下ret = EAGAIN(0) ， 说明数据已经读取完毕
                if (ret == EAGAIN)
                {
                    ELOG("The data has been read");
                    return -1;
                }
                else if (ret == EINTR)
                {
                    ELOG("IO abort");
                    return -1;
                }
                else
                {
                    ELOG("Unknow ERR");
                    return -1;
                }
            }
            ILOG("读取了 %d 个字节",ret);
            return ret;
        }

        // 非阻塞读取
        ssize_t NonBlockRecv(void *buf, size_t len)
        {
            if(len == 0) return 0;
            return Recv(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT  是非阻塞选项
        }

        ssize_t Send(const void *buf, size_t len, int flag = 0)
        {
            int ret = send(_sock, buf, len, flag);
            // 阻塞模式下ret = 0 说明对端连接关闭
            if (ret < 0)
            {
                // 非阻塞模式下ret = EAGAIN(0) ， 说明数据已经读取完毕
                if (ret == EAGAIN)
                {
                    ELOG("The data has been writed");
                    return -1;
                }
                else if (ret == EINTR)
                {
                    ELOG("IO abort");
                    return -1;
                }
                else
                {
                    ELOG("Unknow ERR");
                    return -1;
                }
            }
            return ret;
        }

        // 非阻塞写入
        ssize_t NonBlockSend(void *buf, size_t len)
        {
            if (len == 0) return 0;
            return Send(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT  是非阻塞选项
        }

        void Close() 
        {
            if (_sock >= 0)
            {
                close(_sock);
                _sock = -1;
            }
        }

        bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool flag_block = false)
        {
            if (CreateSocket() && Bind(ip, port))
            {
                ReuserAdder();
                if(!Listen()) return false;
                // false为非阻塞等待，true为阻塞等待
                if (flag_block)
                    NonBlock();
                return true;
            }
            DLOG("create server failed");
            return false;
        }
        bool CreateClient(uint16_t port, const std::string &ip)
        {
            if (CreateSocket() && Connect(ip, port))
            {
                return true;
            }
            DLOG("create client failed");
            return false;
        }

        void ReuserAdder()
        {
            int val = 1;
            int ret = setsockopt(_sock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &val, sizeof(val));
            if (ret < 0)
            {
                DLOG("set reuseaddr and reuseport failed");
            }
        }

        // 设置文件描述符非阻塞
        void NonBlock()
        {
            int flag = fcntl(_sock, F_GETFL);
            if (flag == -1)
            {
                ELOG("set NonBlock failed");
                return;
            }
            fcntl(_sock, F_SETFL, flag | O_NONBLOCK);
        }

    private:
        int _sock;
    };

    // 事件管理类
    class Poller;
    class EventLoop;
    class Channel
    {
    private:
        int _fd;
        EventLoop *_eloop;
        uint32_t _events;  // 需要监控的事件
        uint32_t _revents; // 连接触发的事件
        using func_t = std::function<void()>;
        func_t _read_cb;  // 可读事件被触发的回调函数
        func_t _write_cb; // 可写事件被触发的回调函数
        func_t _error_cb; // 错误事件被触发的回调函数
        func_t _close_cb; // 断开连接事件被触发的回调函数
        func_t _event_cb; // 任意事件被触发的回调函数
    public:
        Channel(int fd, EventLoop *eloop) : _fd(fd), _eloop(eloop), _events(0), _revents(0){};
        int GetFd() { return _fd; }
        uint32_t GetEvent() { return _events; }
        void SetREvents(uint32_t events) { _revents = events; }

        // 设置读事件触发的回调函数
        void SetReadCallBack(const func_t cb) { _read_cb = cb; }
        // 设置写事件触发的回调函数
        void SetWriteCallBack(const func_t cb) { _write_cb = cb; }
        // 设置错误事件触发的回调函数
        void SetErrorCallBack(const func_t cb) { _error_cb = cb; }
        // 设置连接关闭事件触发的回调函数
        void SetCloseCallBack(const func_t cb) { _close_cb = cb; }
        // 设置任意事件的回调函数
        void SetEventCallBack(const func_t cb) { _event_cb = cb; }

        // 读事件监控是否开启
        bool ReadAble() { return _events & EPOLLIN; }
        // 写事件监控是否开启
        bool WriteAble() { return _events & EPOLLOUT; }
        // 开启/关闭 读事件监控
        void EnableRead(bool flag)
        {
            // flag 为true则开启，为false则关闭
            if (flag)
                _events |= EPOLLIN;
            else
                _events &= ~EPOLLIN;
            Mod();
        }
        // 开启/关闭 写事件监控
        void EnableWrite(bool flag)
        {
            // flag 为true则开启，为false则关闭
            if (flag)
                _events |= EPOLLOUT;
            else
                _events &= ~EPOLLOUT;
            Mod();
        }
        // 关闭所有事件监控
        void DisableAll()
        {
            _events = 0;
            Mod();
        }
        void Mod(); // 更改事件关心
        void Del(); // 移除监控(从epoll里的红黑树中移除)

        // 处理事件(根据事件调用对应的回调函数)
        void HandleEvent()
        {
            if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
            {
                // EPOLLRDHUP是对端关闭处理，EPOLLPRI是优先级
                if (_read_cb)
                {
                    ILOG("调用读处理函数");
                    _read_cb();
                }   
            }
            if (_revents & EPOLLOUT)
            {
                if (_write_cb)
                    _write_cb();
            }
            else if (_revents & EPOLLERR)
            {
                if (_error_cb)
                    _error_cb();
            }
            else if (_revents & EPOLLHUP) // EPOLLHUP关闭连接事件处理
            {
                if (_close_cb)
                    _close_cb();
            }
            if (_event_cb)
               _event_cb();
        }
    };

// Epoll模块
#define MAX_EPOLLEVENTS 1024
    class Poller
    {
    private:
        int _epfd;
        struct epoll_event _evts[1024]; // 存储就绪事件的数组
        std::unordered_map<int, Channel *> _channels;

    private:
        void CtlEpoll(Channel *channel, int op)
        {
            struct epoll_event evt;
            evt.data.fd = channel->GetFd();
            evt.events = channel->GetEvent();
            int ret = epoll_ctl(_epfd, op, evt.data.fd, &evt);
            if (ret != 0)
            {
                ELOG("EPOLL_CTL FAILED");
            }
        }
        bool HasChannel(Channel *channel)
        {
            auto it = _channels.find(channel->GetFd());
            return !(it == _channels.end());
        }

    public:
        Poller()
        {
            _epfd = epoll_create(MAX_EPOLLEVENTS);
            if (_epfd < 0)
            {
                ELOG("epoll create failed");
                exit(1);
            }
        }
        ~Poller() { close(_epfd); }

        void ModEvent(Channel *channel)
        {
            if (HasChannel(channel))
                CtlEpoll(channel, EPOLL_CTL_MOD);
            else
            {
                _channels.insert(std::make_pair(channel->GetFd(), channel));
                CtlEpoll(channel, EPOLL_CTL_ADD);
            }
        }
        void DelEvent(Channel *channel)
        {
            auto it = _channels.find(channel->GetFd());
            if (it != _channels.end())
            {
                _channels.erase(it);
            }
            CtlEpoll(channel, EPOLL_CTL_DEL);
        }
        // 开始监控，返回活跃连接
        void WaitEpoll(std::vector<Channel *> *active)
        {
            int n = epoll_wait(_epfd, _evts, MAX_EPOLLEVENTS, -1);
            if (n < 0)
            {
                if (errno == EINTR)
                {
                    DLOG("IO ABORT");
                    return;
                }
                if (errno == EAGAIN)
                {
                    DLOG("data has been read");
                    return;
                }
                ELOG("EPOLL WAIT ERROR : %s", strerror(errno));
            }
            for (int i = 0; i < n; i++)
            {
                auto it = _channels.find(_evts[i].data.fd);
                assert(it != _channels.end());
                it->second->SetREvents(_evts[i].events); // 设置已就绪事件
                active->push_back(it->second);           // 把当前事件管理添加到数组
            }
        }
    };

    using TaskFunc = std::function<void()>;   // 任务处理函数
    using ReleasFunc = std::function<void()>; // 移除定时器函数

    // 定时任务
    class TimeTask
    {
    private:
        uint64_t _id;    // 定时任务的id
        uint32_t _delay; // 推迟的秒数
        TaskFunc _task_cb;
        ReleasFunc _releas_cb;
        bool _isCancle;

    public:
        TimeTask(uint64_t id, uint32_t delay, const TaskFunc &cb) : _id(id), _delay(delay), _task_cb(cb), _isCancle(false) {}

        // 获取定时任务的延迟时间
        uint32_t GetDelayTime() { return _delay; }

        // 设置定时器的移除任务
        void SetReleas(const ReleasFunc &cb) { _releas_cb = cb; }

        // 取消定时任务
        void Cancle() { _isCancle = true; }

        ~TimeTask()
        {
            if (_isCancle == false)
                _task_cb(); // 处理定时器任务
            _releas_cb();   // 销毁定时器任务
        }
    };

    // 时间轮
    class TimerWheel
    {
    private:
        using WeakTask = std::weak_ptr<TimeTask>; // 用weak_ptr是为了防止shared_ptr管理原始对象时出现引用计数重置的情况
        using SharedTask = std::shared_ptr<TimeTask>;
        int _tick;                                      // 时间轮秒表
        int _capacity;                                  // 时间轮的容量
        std::vector<std::vector<SharedTask>> _wheel;    // 时间轮
        std::unordered_map<uint64_t, WeakTask> _timers; // 定时任务ID和初始任务的映射
        
        int _timerfd; //定时文件描述符
        EventLoop* _eloop; 
        std::unique_ptr<Channel> _channel;
    private:

        void RemoveTimer(uint64_t id) // 移除定时器任务对象
        {
            auto it = _timers.find(id);
            if (it != _timers.end())
            {
                _timers.erase(it);
            }
        }

        //创建timerfd
        static int CreateTimerfd()
        {
            int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
            if (timerfd < 0)
            {
                perror("timerfd create failed");
                abort();
            }
            struct itimerspec itp;
            itp.it_value.tv_sec = 1; // 第一次触发的时间
            itp.it_value.tv_nsec = 0;
            itp.it_interval.tv_sec = 1; // 第一次超时后每隔多少秒触发一次
            itp.it_interval.tv_nsec = 0;
            timerfd_settime(timerfd, 0, &itp, nullptr);
            return timerfd;
        }


        //添加定时器
        void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb)
        {
            SharedTask sptr(new TimeTask(id, delay, cb));                   // 创建一个定时任务对象
            sptr->SetReleas(std::bind(&TimerWheel::RemoveTimer, this, id)); // 设置wptr的移除定时任务处理函数
            int pos = (_tick + delay) % _capacity;                          // 秒表位置，环形结构
            _wheel[pos].push_back(sptr);                                    // 把任务加进时间轮
            _timers[id] = WeakTask(sptr);                                   // 建立智能指针和i任务d的映射关系
        }

        //刷新定时器
        void TimerRefreshInLoop(uint64_t id)
        {
            // 找到id对应的智能指针
            auto it = _timers.find(id);
            if (it == _timers.end())
            {
                std::cout << "没有找到定时任务，无法刷新 " << std::endl;
                return;
            }
            // 通过weak_ptr构造出一个shared_ptr对象
            SharedTask sptr = it->second.lock();
            int delay = sptr->GetDelayTime(); // 获取延迟时间

            // 加入到时间轮中
            int pos = (_tick + delay) % _capacity;
            _wheel[pos].push_back(sptr);
            std::cout << "刷新定时任务" << pos << std::endl;
        }
        //取消定时器
        void TimerCancleInLoop(uint64_t id)
        {
            // 找到id对应的智能指针
            auto it = _timers.find(id);
            if (it == _timers.end())
            {
                DLOG("定时器未找到，结束....");
                return;
            }
            SharedTask sptr = it->second.lock();
            if(sptr)
              sptr->Cancle();
        }

        //读取超时次数
        int ReadTimefd()
        {
            uint64_t times = 0;
            //int n = recv(_timerfd,&times,8,MSG_WAITALL);
            int n = read(_timerfd,&times,sizeof(times));
            if(n < 0)
            {
                ELOG("timerfd read failed");
                abort();
            }
            return times; 
        }

        // 时间轮运行
        void RunTimerWheel()
        {
            _tick = (_tick + 1) % _capacity;
            _wheel[_tick].clear();
        }

        //timerfd绑定的读事件
        void OnTime()
        {
            int time = ReadTimefd(); //读取超时次数
            for(int i = 0 ; i < time ; i ++)
            {
                RunTimerWheel(); //运行时间轮
            }
        }

    public:
        TimerWheel(EventLoop* eloop) : _tick(0), _capacity(60), _wheel(60),_eloop(eloop),_timerfd(CreateTimerfd())
                    ,_channel(new Channel(_timerfd,_eloop)) {
                        _channel->SetReadCallBack(std::bind(&TimerWheel::OnTime,this));
                        _channel->EnableRead(true);
                    } 

        // 添加定时器任务
        void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);
        // 刷新定时任务
         void TimerRefresh(uint64_t id);
        // 取消定时任务
        void TimerCancle(uint64_t id);
        bool HasTimer(uint64_t id)
        {
            auto it = _timers.find(id);
            if(it == _timers.end()) return false;
            return true;
        }
    };

    class EventLoop
    {
    private:
        using Functor = std::function<void()>;
        std::thread::id _thread_id; // 线程id
        int _event_fd;
        Poller _poller;
        std::vector<Functor> _tasks; // 任务池
        std::mutex _mutex;           // 锁
        std::unique_ptr<Channel> _event_channel;
        TimerWheel _timer_wheel; 
    public:
        void RunAllTask()
        {
            std::vector<Functor> functors;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _tasks.swap(functors);
            }
            for (auto &f : functors)
            {
                ILOG("当前线程开水执行任务");
                f();
            }
        }

        static int CreateEventfd()
        {
            int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
            if (efd < 0)
            {
                ELOG("create eventfd failed");
                abort();
            }
            return efd;
        }
        void ReadEventfd()
        {
            uint64_t res = 0;
            int n = read(_event_fd, &res, sizeof(res));
            if (n < 0)
            {
                if (errno == EAGAIN || errno == EINTR)
                    return;
                ELOG("recv failed");
                abort();
            }
        }
        void WeakupEventfd()
        {
            // 写入一个数字唤醒event_fd的读事件
            uint64_t val = 1;
            int n = write(_event_fd, &val, sizeof(val));
            if (n <= 0)
            {
                if (errno == EINTR)
                    return;
                ELOG("wiret failed");
                abort();
            }
        }

    public:
        EventLoop() : _thread_id(std::this_thread::get_id()), _event_fd(CreateEventfd()), 
                        _event_channel(new Channel(_event_fd, this)),
                        _timer_wheel(this)
        {
            // 绑定读时间
            ILOG("正在创建EventLoop.....");
            _event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEventfd, this));
            _event_channel->EnableRead(true); // 启动读事件关心
            ILOG("创建EventLoop成功!");
        }

        // 事件监控->就绪处理->执行任务
        void Start()
        {
            while(1)
            {
              // 事件监控
               std::vector<Channel *> actives;
               _poller.WaitEpoll(&actives);

            // 事件处理
               for (auto &e : actives)
                e->HandleEvent();

            // 执行任务
               RunAllTask();
           }
        }
        // 判断要执行的任务是否在当前线程中，是就执行不是就压入队列
        void RunInLoop(const Functor &task)
        {
            if (IsInLoop())
                task();
            else
                QueueInLoop(task);
            
        }

        // 把要执行的任务压入队列
        void QueueInLoop(const Functor &task)
        {
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _tasks.push_back(task);
            }
            // 唤醒Epoll
            WeakupEventfd();
        }
        // 判断当前线程是否是EventLoop对应的线程
        bool IsInLoop() { return _thread_id == std::this_thread::get_id(); }
        void AssertInLoop() {assert(_thread_id == std::this_thread::get_id()); }
        
        void ModEvent(Channel *channel) { _poller.ModEvent(channel); }
        void DelEvent(Channel *channel) { _poller.DelEvent(channel); }

        //TimeWheel控制
        void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb){_timer_wheel.TimerAdd(id,delay,cb);}
        void TimerRefresh(uint64_t id){_timer_wheel.TimerRefresh(id);}
        void TimerCancle(uint64_t id){_timer_wheel.TimerCancle(id);}
        bool HasTimer(uint64_t id){return _timer_wheel.HasTimer(id);}
    };
    
    
    class ThreadLoop
    {
    private:
      std::mutex _mutex; 
      std::condition_variable _cond;  
      EventLoop *_loop;
      std::thread _thread; 
    
    private:
      void ThreadEntry()
      {
        EventLoop loop;
        {
          std::unique_lock<std::mutex> lock(_mutex);
          _loop = &loop;
          _cond.notify_all();
        }
        ILOG("loop Start!");
        loop.Start();
      }
    public:
      ThreadLoop():_loop(nullptr),_thread(std::thread(&ThreadLoop::ThreadEntry,this)){}

      EventLoop* GetLoop()
      {
        EventLoop* loop = nullptr; 
        {
          std::unique_lock<std::mutex> lock(_mutex);
          _cond.wait(lock,[&](){return _loop != nullptr;});
          loop = _loop;
        }
        if(loop == nullptr) ILOG("loop == nullptr");
        ILOG("GetLoop success,");
        return loop;
      }
    };
  
    
    class ThreadLoopPool
    {
      private:
        int _thread_count; 
        int _next_id;
        EventLoop* _base_loop; 
        std::vector<EventLoop*> _loops;
        std::vector<ThreadLoop*> _threads;

      public:
        ThreadLoopPool(EventLoop* base_loop) : _thread_count(0),_next_id(0), _base_loop(base_loop){}
        
        void SetCount(int count) {_thread_count = count; }
        
        void InitThreadLoop()
        {
          if(_thread_count > 0)
          {
            _threads.resize(_thread_count); 
            _loops.resize(_thread_count); 
            for(int i = 0 ;  i < _thread_count ; i++)
            {
              _threads[i] = new ThreadLoop();
              _loops[i] = _threads[i]->GetLoop();
            }
          }
        }

        EventLoop* NextLoop()
        {
          if(_thread_count == 0) return _base_loop; 
          _next_id = (_next_id + 1) % _thread_count;
          return _loops[_next_id];
        }
    };
    class Any
    {
    private:
        // 抽象类
        class holder
        {
        public:
            virtual ~holder() {}
            virtual holder* clone() = 0;
            virtual const std::type_info &type() = 0;
        };

        template <class T>
        class placeholder : public holder
        {
        public:
            placeholder(const T &val) : _val(val) {}
            virtual holder* clone() { return new placeholder<T>(_val); }
            virtual const std::type_info &type() { return typeid(T); }

        public:
            T _val;
        };

    private:
        holder *_content;

    public:
        Any() :_content(nullptr){}
        ~Any(){ delete _content;} 

        template <class T>
        Any(const T& val) : _content(new placeholder<T>(val)) {}

        Any(const Any &other):_content(other._content ? other._content->clone() : nullptr) {}

        template <class T>
        Any& operator=(const T &val) {
            Any(val).swap(*this);
            return *this;
        }
        Any& operator=(const Any &other) {
            Any(other).swap(*this);
            return *this;
        }
    public:
        void swap(Any &other) {
            std::swap(_content,other._content);
        }
        template <class T>
        T *get()
        {
            assert(typeid(T) == _content->type());
            return &((placeholder<T>*)_content)->_val;
        }
    };

    class Connection;
    using PtrConnection = std::shared_ptr<Connection>;
    class Connection : public std::enable_shared_from_this<Connection>
    {
      typedef enum {
        DISCONNECTED, //连接关闭状态
        CONNECTING,  //连接建立成功，待处理状态
        CONNECTED ,  //连接建立完成，可以通信状态
        DISCONNECTING //连接等待关闭状态
      }ConnStatu;

    private:
        uint64_t _con_id; //连接ID
        int _sockfd; //管理的套接字描述符
        bool _is_inactive_release;  //是否启动非活跃连接销毁 
        ConnStatu _statu; //连接状态
        EventLoop* _loop;
        Socket _socket;  //连接的套接字对象 
        Channel _channel;   //对连接的事件管理
        Buffer _in_buff;  //输入缓冲区 
        Buffer _out_buff; //输出缓冲区
        Any _context;    //请求的接收处理上下文 

        using ConnectedCallBack = std::function<void(const PtrConnection&)>;
        using MessageCallBack = std::function<void(const PtrConnection&,Buffer*)>;
        using ClosedCallBack = std::function<void(const PtrConnection&)>;
        using AnyEventCallBack = std::function<void(const PtrConnection&)>;
        ConnectedCallBack _con_cb;      //
        MessageCallBack _msg_cb;        //消息处理回调函数
        ClosedCallBack _cls_cb;         //连接关闭回调函数(用户使用)
        AnyEventCallBack _any_event_cb; //任意事件触发回调函数

        ClosedCallBack _server_close_cb;  //服务器有连接关闭时使用
    private:  
        //五个Channel事件处理回调函数
        void HandleRead()
        {
          char buf[65536] = {0};
          ILOG("进入非阻塞读取 : %s",buf);
          int ret = _socket.NonBlockRecv(buf,65535);
          if(ret < 0)
          {
            DLOG("连接 %d 读取失败，已关闭。",_sockfd); 
            ShutdownInLoop();
            return; 
          }
          ILOG("读取完毕");
          if(ret == 0) return; //没读取到数据 
          _in_buff.WriteAndPush(buf,ret);
          if(_in_buff.ReadAbleSize() > 0)
          {
            _msg_cb(shared_from_this(),&_in_buff); //调用消息处理函数
          }
        }
        void HandleWrite()
        {
          //把输出缓冲区的数据用非阻塞的方式发送
          int ret = _socket.NonBlockSend(_out_buff.ReadPosition(),_out_buff.ReadAbleSize()); 
          //发送失败
          if(ret < 0)
          {
            //如果输入缓冲区中还有数组，那么关闭前先发送输入缓冲区中的数据
            if(_in_buff.ReadAbleSize() > 0)
            {
              _msg_cb(shared_from_this(),&_in_buff);
            }
            //释放连接
            return Release();
          }
          //因为前面的数据已经发送，所以读数据往后偏移
          _out_buff.MoveReadOffset(ret); 
          
          //如果输出缓冲区没有数据
          if(_out_buff.ReadAbleSize() == 0)
          {
            _channel.EnableWrite(false); //关闭对写事件的关心
            
            //如果连接状态为待关闭状态，那么释放连接
            if(_statu == DISCONNECTING) Release();
          }
        }
        void HandleClose()
        {
            DLOG("开始关闭连接");
            //如果输入缓冲区中还有数组，那么关闭前先发送输入缓冲区中的数据
            if(_in_buff.ReadAbleSize() > 0)
            {
              _msg_cb(shared_from_this(),&_in_buff);
            }
            //释放连接
            return Release();
        }
        void HandleError()
        {
          HandleClose();
        }
        void HandleEvent()
        {
          //如果启用了非活跃连接销毁，那么刷新超时时间
          if(_is_inactive_release == true) _loop->TimerRefresh(_con_id);
          if(_any_event_cb) _any_event_cb(shared_from_this());
        }
        
        //连接获取完毕后对连接的事件处理函数进行设置
        void EstablishedInLoop()
        {
          assert(_statu == CONNECTING); 
          _statu = CONNECTED;
          _channel.EnableRead(true); 
          if(_con_cb) _con_cb(shared_from_this());
        }
        
        //真正的关闭连接接口
        void ReleaseInLoop()
        {
          //1.修改连接状态
          _statu = DISCONNECTED; 
          ILOG("修改连接状态%d",_statu);
          //2移除事件监控
          _channel.Del();
          ILOG("移除事件监控");
          //3.关闭描述符 
          _socket.Close();
          ILOG("关闭文件描述符");
          //4.销毁定时器任务
          if(_loop->HasTimer(_con_id))
          {
            ILOG("正在销毁定时器任务");
            EnableInactiveReleaseInLoop(false);
          }
          //5.调用关闭回调函数
          if(_cls_cb){ ILOG("调用用户设置的关闭函数"); _cls_cb(shared_from_this()); }//用户自己设置的关闭函数
          if(_server_close_cb){ ILOG("调用服务器的连接关闭函数"); _server_close_cb(shared_from_this());} //服务器本身的关闭函数
        }
          
        void Release()
        {
          _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
        }

        void SendInLoop(Buffer& buf)
        {
          
          if(_statu == DISCONNECTED) return; //连接已经关闭，无法发送
          //把要发送的数据拷贝到输出缓冲区，并且开启写事件  
          _out_buff.WriteBufferAndPush(buf);
          if(!_channel.WriteAble())
          { 
            _channel.EnableWrite(true); //写事件没有开启则开启写事件
          } 
        }

        void ShutdownInLoop()
        {
          _statu = DISCONNECTING; //设置连接为即将关闭状态 
          //如果输入缓冲区有数据，读进来
          if(_in_buff.ReadAbleSize() > 0)
          {
            if(_msg_cb) _msg_cb(shared_from_this(),&_in_buff);
          }
          
          //如果输出缓冲区有数据，写出去
          if(_out_buff.ReadAbleSize() > 0)
          {
            if(!_channel.WriteAble()) _channel.EnableWrite(true); //写事件没开启则开启写事件
          }
          
          //输出缓冲区没有数据了
          if(_out_buff.ReadAbleSize() == 0) Release();

        }

        void EnableInactiveReleaseInLoop(bool flag,int sec = -1)
        {
          _is_inactive_release = flag;
          if(flag)
          {
            //启用非活跃连接销毁
            //如果任务已经存在，那么刷新，不存在则新增 
            DLOG("启用非活跃连接销毁");
            if(_loop->HasTimer(_con_id))return  _loop->TimerRefresh(_con_id); 
       
            else{ILOG("定时器不存在，添加"); _loop->TimerAdd(_con_id,sec,std::bind(&Connection::Release,this));
            }}else 
          {
            //关闭非活跃连接销毁
            //如果连接存在，则关闭
            ILOG("关闭非活跃连接销毁");
            if(_loop->HasTimer(_con_id))
            {
              ILOG("正在销毁定时器任务.........%d",_con_id);
              _loop->TimerCancle(_con_id); 

            }  
          }
         
      }
       void UpgradeInLoop(const Any& context,
                        const ConnectedCallBack& con , 
                        const MessageCallBack& msg,
                        const ClosedCallBack& cls,
                        const AnyEventCallBack any_event)
       {
         SetContext(context);
         SetClosedCallBack(cls);
         SetConnectedCallBack(con);
         SetMessageCallBack(msg);
         SetAnyEventCallBack(any_event);
       }

    public:
       Connection(EventLoop* loop,uint64_t id , int sockfd):_con_id(id),_sockfd(sockfd),
                _is_inactive_release(false) , _statu(CONNECTING),_loop(loop),_socket(_sockfd),_channel(_sockfd,_loop)
       {
          //设置回调函数 
          _channel.SetReadCallBack(std::bind(&Connection::HandleRead,this));
          _channel.SetWriteCallBack(std::bind(&Connection::HandleWrite,this));
          _channel.SetCloseCallBack(std::bind(&Connection::HandleClose,this));
          _channel.SetErrorCallBack(std::bind(&Connection::HandleError,this));
          _channel.SetEventCallBack(std::bind(&Connection::HandleEvent,this));
       }
       ~Connection(){DLOG("连接%p已经释放",this);} 
      
       //获取fd
       int Fd() {return _sockfd;}
      
       //获取连接id
       int Id() {return _con_id;}
      
       //是否处于CONNETED状态
       bool Connected(){return _statu == CONNECTED;} 
      
       //设置上下文
       void SetContext(const Any& context) {_context = context;}
        
       //获取上下文
       Any* GetContext() {return &_context;}

       void SetConnectedCallBack(const ConnectedCallBack& cb){_con_cb = cb;}
       void SetMessageCallBack(const MessageCallBack& cb) { _msg_cb = cb; }
       void SetClosedCallBack(const ClosedCallBack& cb){ _cls_cb = cb; }
       void SetAnyEventCallBack(const AnyEventCallBack& cb){ _any_event_cb = cb; }
       void SetSvrClosedCallBack(const ClosedCallBack& cb){ _server_close_cb = cb; }

       //建立连接后设置回调函数，启动读事件
       void Established(){_loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));} 

       //数据发送
       void Send(const char* data,size_t len){
            Buffer buf; 
            buf.WriteAndPush(data,len);
            ILOG("%s",buf.ReadPosition());
            _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buf)));
        } 
      
       //关闭连接
       void Shutdown(){_loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));}
       //启用/关闭非活跃连接销毁
       void EnableInactiveRelease(bool flag , int sec= -1){
         _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,flag,sec));}
      
       //协议切换
       void Upgrade(const Any& context,const ConnectedCallBack& con , const MessageCallBack& msg
                    ,const ClosedCallBack& cls,const AnyEventCallBack any_event)
       {
          _loop->AssertInLoop(); 
          _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,con,msg,cls,any_event));
       }
    };
    
    class Acceptor 
    {
    private:
      Socket _socket; 
      EventLoop* _eloop; 
      Channel _channel; 

      using AcceptCallBack = std::function<void(int)>; 
      AcceptCallBack _a_cb;
      
      void HandleRead()
      {
        int new_fd = _socket.Accept();
        if(new_fd < 0)
        {
          ELOG("Accept sock failed");
          return;
        }
        if(_a_cb) _a_cb(new_fd);
      }

      int CreateServer(uint16_t port)
      {
        bool ret = _socket.CreateServer(port);
        if(!ret)
        {
          DLOG("CreateServer failed");
          abort();
        }
        return _socket.GetFd();
      }

    public:
      Acceptor(EventLoop* eloop, uint16_t port): _socket(CreateServer(port)),_eloop(eloop),_channel(Channel(_socket.GetFd(),_eloop))
      {
        _channel.SetReadCallBack(std::bind(&Acceptor::HandleRead,this));
      }
      void Listen(){_channel.EnableRead(true); /*开启读事件关心*/}
      ~Acceptor(){} 
      void SetAcceprCallBack(const AcceptCallBack& cb){_a_cb = cb;}
    
    };
    
    class TcpServer
    {
    private: 
        uint16_t _port;
        uint64_t _next_id; 
        int _timeout; 
        bool _enable_inactive_release; 
        Acceptor _acceptor; 
        EventLoop _baseloop; 
        ThreadLoopPool _pool;
        std::unordered_map<uint64_t,PtrConnection> _conns;
      
        using Functor = std::function<void()>;
        using ConnectedCallBack = std::function<void(const PtrConnection&)>;
        using MessageCallBack = std::function<void(const PtrConnection&,Buffer*)>;
        using ClosedCallBack = std::function<void(const PtrConnection&)>;
        using AnyEventCallBack = std::function<void(const PtrConnection&)>;
        ConnectedCallBack _con_cb;      //
        MessageCallBack _msg_cb;        //消息处理回调函数
        ClosedCallBack _cls_cb;         //连接关闭回调函数(用户使用)
        AnyEventCallBack _any_event_cb; //任意事件触发回调函数
    private:
        void NewConnection(int fd)
        {
            _next_id++;
            PtrConnection conn(new Connection(_pool.NextLoop(),_next_id,fd));
            conn->SetMessageCallBack(_msg_cb);
            conn->SetConnectedCallBack(_con_cb);
            conn->SetClosedCallBack(_cls_cb);
            conn->SetAnyEventCallBack(_any_event_cb);
            conn->SetSvrClosedCallBack(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
            if(_enable_inactive_release) conn->EnableInactiveRelease(true,_timeout);
            conn->Established();
            _conns.insert(std::make_pair(_next_id,conn));
            DLOG("NEW----------------------------");
        }
        void RemoveConnection(const PtrConnection conn)
        {
          int id = conn->Id(); 
          auto it = _conns.find(id);
          if(it != _conns.end()) _conns.erase(it);
        }

        void RunAfterInLoop(const Functor& task,int delay)
        {
          _next_id++; 
          _baseloop.TimerAdd(_next_id,delay,task);
        }
    public:
        TcpServer(uint16_t port):_port(port),_next_id(0),_enable_inactive_release(false),_acceptor(&_baseloop,_port),_pool(&_baseloop){
          _acceptor.SetAcceprCallBack(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
          _acceptor.Listen(); //开始监听
        }
        void SetConnectedCallBack(const ConnectedCallBack& cb){_con_cb = cb;}
        void SetMessageCallBack(const MessageCallBack& cb) { _msg_cb = cb; }
        void SetClosedCallBack(const ClosedCallBack& cb){ _cls_cb = cb; }
        void SetAnyEventCallBack(const AnyEventCallBack& cb){ _any_event_cb = cb; }
        
        void SetThreadCount(int count) {_pool.SetCount(count);}
        void EnableInactiveRelease(int timeout){_timeout = timeout; _enable_inactive_release = true;}
        void RunAfter(const Functor& task,int delay)
        {
          _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
        }
        void Start()
        {
          _pool.InitThreadLoop(); //初始化线程池
          _baseloop.Start();
        }
    };

    void Channel::Mod() { _eloop->ModEvent(this); }
    void Channel::Del() { _eloop->DelEvent(this); }
    void TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
    { 
        _eloop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,cb));
     }

    // 刷新定时任务
    void TimerWheel::TimerRefresh(uint64_t id)
    {
        _eloop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
    }

    // 取消定时任务
    void TimerWheel::TimerCancle(uint64_t id)
    {
        _eloop->RunInLoop(std::bind(&TimerWheel::TimerCancleInLoop,this,id));
    }

}
